feat(fullnode) Add filterOrders option to streaming subscription #2676
Conversation
WalkthroughThis pull request introduces a new boolean parameter Changes
Possibly related PRs
Suggested labels
Suggested reviewers
Poem
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (7)
protocol/streaming/util/util.go (2)
25-26: Improve function documentation.The current comment is focused on error cases. Consider adding a description of the function's purpose and return values.
-// Error expected if OffChainUpdateV1.UpdateMessage message type is extended to more order events +// GetOffChainUpdateV1SubaccountIdNumber extracts the SubaccountId.Number from an OffChainUpdateV1 message. +// Returns the subaccount number and an error if the update message type is not a supported order event +// (OrderPlace, OrderRemove, OrderUpdate, OrderReplace).
28-42: Consider using constants for error messages.The error message is hardcoded. Consider defining it as a constant to maintain consistency and make updates easier.
+const ( + ErrUnsupportedUpdateMessage = "UpdateMessage type not in {OrderPlace, OrderRemove, OrderUpdate, OrderReplace}: %+v" +) switch updateMessage := update.UpdateMessage.(type) { // ... case statements ... default: - return 0, fmt.Errorf( - "UpdateMessage type not in {OrderPlace, OrderRemove, OrderUpdate, OrderReplace}: %+v", - updateMessage, - ) + return 0, fmt.Errorf(ErrUnsupportedUpdateMessage, updateMessage)protocol/streaming/util/util_test.go (2)
113-113: Remove debug print statement.The
fmt.Printlnstatement appears to be left from debugging.- fmt.Println("expected", id)
21-47: Refactor test data setup.Consider moving the test data setup into helper functions to improve readability and reusability.
func createTestOrder(subaccountIdNumber uint32) (pv1types.IndexerOrderId, pv1types.IndexerOrder) { orderId := pv1types.IndexerOrderId{ SubaccountId: pv1types.IndexerSubaccountId{ Owner: "foo", Number: subaccountIdNumber, }, ClientId: 0, OrderFlags: 0, ClobPairId: 0, } order := pv1types.IndexerOrder{ OrderId: orderId, // ... other fields ... } return orderId, order }protocol/streaming/full_node_streaming_manager.go (1)
223-277: Ensure comprehensive error handling and future-proofing for additional update typesIn
FilterSubaccountStreamUpdates, whenGetOffChainUpdateV1SubaccountIdNumberreturns an error, the error is logged but not addressed further. Consider whether logging alone is sufficient or if additional error handling is necessary to prevent potential data loss of critical updates.Moreover, currently, only
StreamUpdate_OrderbookUpdatemessages are explicitly filtered. If new update message types containing subaccount IDs are introduced in the future, they might bypass filtering. Consider extending the filtering logic to handle other relevant message types to ensure consistent behavior.protocol/streaming/types/interface.go (1)
19-19: Add documentation for the newfilterOrdersparameterIncluding comments to describe the purpose and usage of the
filterOrdersparameter in theSubscribemethod will improve code readability and assist developers in understanding the subscription behavior.Apply this diff to add documentation:
Subscribe( clobPairIds []uint32, subaccountIds []*satypes.SubaccountId, marketIds []uint32, + // filterOrders determines whether to filter order updates based on subaccountIds filterOrders bool, srv OutgoingMessageSender, ) ( err error, )proto/dydxprotocol/clob/query.proto (1)
189-191: Add detailed documentation for the filter_orders field.The new field would benefit from more comprehensive documentation explaining:
- Its purpose and behavior
- The default value implications
- The relationship with subaccount filtering
Add documentation comments above the field:
// Market ids for price updates. repeated uint32 market_ids = 3; + // Indicates whether to filter order updates based on the specified subaccount_ids. + // When true, only order updates related to the specified subaccount_ids will be streamed. + // When false or unspecified, all order updates will be streamed regardless of subaccount_ids. + // This filter works in conjunction with position update filtering specified by subaccount_ids. bool filter_orders = 4;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
protocol/x/clob/types/query.pb.gois excluded by!**/*.pb.go
📒 Files selected for processing (10)
indexer/packages/v4-protos/src/codegen/dydxprotocol/clob/query.ts(6 hunks)proto/dydxprotocol/clob/query.proto(1 hunks)protocol/streaming/full_node_streaming_manager.go(6 hunks)protocol/streaming/full_node_streaming_manager_test.go(1 hunks)protocol/streaming/noop_streaming_manager.go(1 hunks)protocol/streaming/types/interface.go(1 hunks)protocol/streaming/util/util.go(1 hunks)protocol/streaming/util/util_test.go(1 hunks)protocol/streaming/ws/websocket_server.go(3 hunks)protocol/x/clob/keeper/grpc_stream_orderbook.go(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (9)
- GitHub Check: test-sim-after-import
- GitHub Check: test-sim-import-export
- GitHub Check: test-sim-multi-seed-short
- GitHub Check: test / run_command
- GitHub Check: unit-end-to-end-and-integration
- GitHub Check: test-race
- GitHub Check: test-coverage-upload
- GitHub Check: container-tests
- GitHub Check: Summary
🔇 Additional comments (14)
protocol/streaming/full_node_streaming_manager.go (3)
97-114: Encapsulation ofOrderbookSubscriptioninitialization improves code reusabilityThe new function
NewOrderbookSubscriptioncleanly encapsulates the initialization logic forOrderbookSubscription, enhancing code reuse and maintainability.
116-131: RefactoredNewOrderbookSubscriptionmethod promotes code simplicityBy delegating to the general
NewOrderbookSubscriptionfunction, the method withinFullNodeStreamingManagerImplsimplifies subscription creation and reduces code duplication.
1182-1187: Correctly caching updates to synchronize local operations queueThe addition of caching
syncLocalUpdatesensures that local operations are properly synchronized duringStreamBatchUpdatesAfterFinalizeBlock, enhancing the reliability and consistency of the streaming service.protocol/x/clob/keeper/grpc_stream_orderbook.go (1)
15-15: Inclusion offilterOrdersparameter enables order filtering in subscriptionsAdding
req.GetFilterOrders()to theSubscribecall correctly integrates thefilterOrdersoption, allowing clients to optionally filter order updates based on their specified criteria.protocol/streaming/noop_streaming_manager.go (1)
27-27: LGTM!The no-op implementation correctly implements the updated interface by adding the new
filterOrdersparameter.protocol/streaming/ws/websocket_server.go (3)
100-108: LGTM! Error handling is consistent with other parameters.The implementation follows the established pattern for parameter parsing and error handling.
183-194: LGTM! Helper function follows best practices.The
parseFilterOrdersfunction:
- Correctly handles empty parameter case
- Uses appropriate error message format
- Returns a default value of
falsewhen parameter is not provided
123-123: LGTM! Parameter correctly passed to Subscribe method.The
filterOrdersparameter is properly passed to the streaming manager's Subscribe method.protocol/streaming/full_node_streaming_manager_test.go (3)
257-261: LGTM! Well-structured test case definition.The
TestCasestruct provides a clear and maintainable way to define test scenarios.
318-454: LGTM! Comprehensive test coverage.The test suite thoroughly covers:
- Base scenarios for filtering
- Edge cases (empty updates, no IDs)
- Different update types (orderbook fills, taker orders, subaccount updates, price updates)
- Multiple subaccount combinations
456-477: LGTM! Robust test execution.The test implementation:
- Properly handles channel cleanup with defer
- Uses appropriate timeouts for no-message scenarios
- Includes clear assertions
indexer/packages/v4-protos/src/codegen/dydxprotocol/clob/query.ts (3)
281-283: LGTM! Clear field documentation.The new field is well-documented with a clear description of its purpose.
1307-1308: LGTM! Proper protobuf encoding implementation.The encoding logic correctly:
- Initializes the field with a default value of
false- Only encodes the field when it's
trueAlso applies to: 1333-1336
1380-1382: LGTM! Proper protobuf decoding implementation.The decoding logic correctly handles the new boolean field.
| tests := map[string]struct { | ||
| update ocutypes.OffChainUpdateV1 | ||
| id uint32 | ||
| err error | ||
| }{ | ||
| "OrderPlace": { | ||
| update: ocutypes.OffChainUpdateV1{ | ||
| UpdateMessage: &ocutypes.OffChainUpdateV1_OrderPlace{ | ||
| OrderPlace: &ocutypes.OrderPlaceV1{ | ||
| Order: &order, | ||
| PlacementStatus: ocutypes.OrderPlaceV1_ORDER_PLACEMENT_STATUS_BEST_EFFORT_OPENED, | ||
| TimeStamp: _ToPtr(orderPlaceTime), | ||
| }, | ||
| }, | ||
| }, | ||
| id: subaccountIdNumber, | ||
| err: nil, | ||
| }, | ||
| "OrderRemove": { | ||
| update: ocutypes.OffChainUpdateV1{ | ||
| UpdateMessage: &ocutypes.OffChainUpdateV1_OrderRemove{ | ||
| OrderRemove: &ocutypes.OrderRemoveV1{ | ||
| RemovedOrderId: &orderId, | ||
| Reason: stypes.OrderRemovalReason_ORDER_REMOVAL_REASON_USER_CANCELED, | ||
| RemovalStatus: ocutypes.OrderRemoveV1_ORDER_REMOVAL_STATUS_CANCELED, | ||
| TimeStamp: _ToPtr(orderPlaceTime.Add(1 * time.Second)), | ||
| }, | ||
| }, | ||
| }, | ||
| id: subaccountIdNumber, | ||
| err: nil, | ||
| }, | ||
| "OrderUpdate": { | ||
| update: ocutypes.OffChainUpdateV1{ | ||
| UpdateMessage: &ocutypes.OffChainUpdateV1_OrderUpdate{ | ||
| OrderUpdate: &ocutypes.OrderUpdateV1{ | ||
| OrderId: &orderId, | ||
| TotalFilledQuantums: fillQuantums, | ||
| }, | ||
| }, | ||
| }, | ||
| id: subaccountIdNumber, | ||
| err: nil, | ||
| }, | ||
| "OrderReplace": { | ||
| update: ocutypes.OffChainUpdateV1{ | ||
| UpdateMessage: &ocutypes.OffChainUpdateV1_OrderReplace{ | ||
| OrderReplace: &ocutypes.OrderReplaceV1{ | ||
| OldOrderId: &orderId, | ||
| Order: &newOrder, | ||
| PlacementStatus: ocutypes.OrderPlaceV1_ORDER_PLACEMENT_STATUS_OPENED, | ||
| TimeStamp: _ToPtr(orderPlaceTime.Add(3 * time.Second)), | ||
| }, | ||
| }, | ||
| }, | ||
| id: subaccountIdNumber, | ||
| err: nil, | ||
| }, |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add negative test cases.
The test suite only covers happy paths. Consider adding test cases for:
- Nil UpdateMessage
- Unsupported message types
- Nil inner objects (Order, OrderId, etc.)
Example additional test cases:
"Nil UpdateMessage": {
update: ocutypes.OffChainUpdateV1{
UpdateMessage: nil,
},
id: 0,
err: fmt.Errorf(ErrUnsupportedUpdateMessage, nil),
},
"Unsupported Message Type": {
update: ocutypes.OffChainUpdateV1{
UpdateMessage: &ocutypes.OffChainUpdateV1_TradeUpdate{},
},
id: 0,
err: fmt.Errorf(ErrUnsupportedUpdateMessage, &ocutypes.OffChainUpdateV1_TradeUpdate{}),
},| // If filterOrders, listen to filtered channel and start filter goroutine | ||
| // Error if filterOrders but no subaccounts are subscribed | ||
| filteredUpdateChannel := subscription.updatesChannel | ||
| if filterOrders { | ||
| if len(subaccountIds) == 0 { | ||
| sm.logger.Error( | ||
| fmt.Sprintf( | ||
| "filterOrders requires subaccountIds for subscription id: %+v", | ||
| subscription.subscriptionId, | ||
| ), | ||
| ) | ||
| } else { | ||
| filteredUpdateChannel = make(chan []clobtypes.StreamUpdate, sm.maxSubscriptionChannelSize) | ||
| defer close(filteredUpdateChannel) | ||
| go subscription.FilterSubaccountStreamUpdates(filteredUpdateChannel, sm.logger) | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
Prevent execution when filterOrders is true without subaccountIds
When filterOrders is true but no subaccountIds are provided, the code logs an error but continues execution. This can lead to subscribers receiving unfiltered order updates, contrary to their request. It's advisable to return an error or terminate the subscription to prevent unintended behavior.
Apply this diff to return an error when filterOrders is requested without subaccountIds:
if filterOrders {
if len(subaccountIds) == 0 {
sm.logger.Error(
fmt.Sprintf(
"filterOrders requires subaccountIds for subscription id: %+v",
subscription.subscriptionId,
),
)
+ return types.ErrInvalidStreamingRequest
} else {
filteredUpdateChannel = make(chan []clobtypes.StreamUpdate, sm.maxSubscriptionChannelSize)
defer close(filteredUpdateChannel)
go subscription.FilterSubaccountStreamUpdates(filteredUpdateChannel, sm.logger)
}
}Committable suggestion skipped: line range outside the PR's diff.
proto/dydxprotocol/clob/query.proto
Outdated
|
|
||
| // Filter order updates in addition to position updates | ||
| bool filter_orders = 4; |
There was a problem hiding this comment.
💡 Codebase verification
Implementation of filter_orders parameter appears incomplete
The filter_orders parameter has been added to the protocol definition and is properly passed through the interface chain, but there's no evidence of actual filtering logic implementation in the codebase. This could lead to the parameter being ineffective.
protocol/streaming/full_node_streaming_manager.go: The streaming manager accepts the parameter but doesn't appear to use it for filtering order updates- Consider implementing the filtering logic in
AddOrderUpdatesToCacheor related methods where updates are processed
🔗 Analysis chain
Verify integration with existing streaming functionality.
The field addition looks correct, but let's verify its integration with the existing streaming system.
Run the following to check for consistent handling across the codebase:
🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash
# Description: Verify the integration of filter_orders field with streaming functionality
# Check for corresponding handler implementations
rg -g '*.{go,proto}' -A 5 'StreamOrderbookUpdates.*Request'
# Check for any existing filtering logic that might need updates
rg -g '*.go' -A 10 'func.*Subscribe'
# Look for tests covering the new functionality
rg -g '*_test.{go,proto}' -A 5 'filter.*orders'
Length of output: 14812
Script:
#!/bin/bash
# Check for filtering logic implementation in streaming manager
rg -g '*.go' -A 15 'filterOrders.*bool' protocol/streaming/
# Look for any tests verifying the filtering functionality
rg -g '*_test.go' -B 5 -A 10 'filterOrders.*true' protocol/streaming/
Length of output: 2484
Script:
#!/bin/bash
# Look for order update streaming implementation
rg -g '*.go' -B 5 -A 10 'SendStreamUpdates|SendOrderbookUpdates' protocol/streaming/
# Check for any filtering logic in the update sending code
rg -g '*.go' -B 5 -A 10 'func.*Update.*Order' protocol/streaming/
# Look for any order-related streaming logic
ast-grep --pattern 'if $filterOrders {
$$$
}'
Length of output: 7169
Script:
#!/bin/bash
# Look for message sending implementation
rg -g '*.go' -B 5 -A 15 'func.*Send.*Message' protocol/streaming/
# Check subscriber management and cache handling
rg -g '*.go' -B 5 -A 15 'AddOrderUpdatesToCache|HandleNewSubscriber' protocol/streaming/
# Look for any references to filter_orders in the implementation
rg -g '*.go' -l 'filter.*orders' protocol/streaming/
Length of output: 6862
| @@ -186,6 +186,9 @@ message StreamOrderbookUpdatesRequest { | |||
|
|
|||
| // Market ids for price updates. | |||
| repeated uint32 market_ids = 3; | |||
|
|
|||
| // Filter order updates in addition to position updates | |||
| bool filter_orders = 4; | |||
There was a problem hiding this comment.
Prefer making it more explicit:
| bool filter_orders = 4; | |
| bool filter_orders_by_subaccount_id = 4; |
| "filterOrders requires subaccountIds for subscription id: %+v", | ||
| subscription.subscriptionId, | ||
| ), | ||
| ) |
There was a problem hiding this comment.
Why not just return error to user and fail the Subscribe request? Then can also remove else below: https://go.dev/wiki/CodeReviewComments#indent-error-flow
| for updates := range sub.updatesChannel { | ||
| filteredUpdates := []clobtypes.StreamUpdate{} | ||
| for _, update := range updates { | ||
| switch updateMessage := update.UpdateMessage.(type) { |
There was a problem hiding this comment.
For modularity, move switch statements into a new function:
func FilterUpdates(input []clobtypes.StreamUpdate) (filteredUpdates []clobtypes.StreamUpdate)
Also makes it easier to test
| for _, orderBookUpdate := range updateMessage.OrderbookUpdate.Updates { | ||
| orderBookUpdateSubaccountIdNumber, err := streaming_util.GetOffChainUpdateV1SubaccountIdNumber(orderBookUpdate) | ||
| if err == nil { | ||
| if slices.Contains(subaccountIdNumbers, orderBookUpdateSubaccountIdNumber) { |
There was a problem hiding this comment.
Any reason to use slice over a map (as a set)? Is it because the former is more efficient at smaller size?
There was a problem hiding this comment.
Yes, it's based on the assumption that the subaccount cardinality is low.
If I had a test that imposed significant load, I could profile both the map and slice.
| } else { | ||
| filteredUpdateChannel = make(chan []clobtypes.StreamUpdate, sm.maxSubscriptionChannelSize) | ||
| defer close(filteredUpdateChannel) | ||
| go subscription.FilterSubaccountStreamUpdates(filteredUpdateChannel, sm.logger) |
There was a problem hiding this comment.
Per offline discussion, let's start with the simpler implementation of doing filtering below in the loop on subscription.updatesChannel, with the assumption that it's acceptable overhead to add an extra boolean comparison on each iteration of the channel.
Co-authored-by: Teddy Ding <[email protected]>
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (1)
proto/dydxprotocol/clob/query.proto (1)
189-192: 🛠️ Refactor suggestionEnhance field name clarity and fix comment formatting.
- The field name should be more explicit about filtering by subaccount ID.
- The comment formatting needs to be adjusted to match clang-format style.
Apply this diff to improve clarity and fix formatting:
- // Filter order updates by subaccount IDs. - // If true, the orderbook updates only include orders from provided subaccount IDs. - bool filter_orders = 4; + // If true, filters order updates to only include orders from the subaccount IDs + // specified in the subaccount_ids field. + bool filter_orders_by_subaccount_id = 4;
🧹 Nitpick comments (1)
proto/dydxprotocol/clob/query.proto (1)
189-192: Enhance documentation for filtering behavior.The documentation should be expanded to clarify:
- The relationship between
subaccount_idsandfilter_orders_by_subaccount_id- The default behavior when the field is not set
- Usage examples
Apply this diff to improve the documentation:
- // Filter order updates by subaccount IDs. - // If true, the orderbook updates only include orders from provided subaccount IDs. - bool filter_orders = 4; + // Controls filtering of order updates based on subaccount_ids field. + // If true: Only order updates from the specified subaccount_ids are included + // If false (default): All order updates are included regardless of subaccount_ids + // Example: Set to true with subaccount_ids=[A,B] to receive updates only for + // orders belonging to subaccounts A and B. + bool filter_orders_by_subaccount_id = 4;
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
proto/dydxprotocol/clob/query.proto(1 hunks)
🧰 Additional context used
🪛 GitHub Actions: Protobuf
proto/dydxprotocol/clob/query.proto
[error] 187-189: Code formatting does not match clang-format style. Line breaks and spacing in comments need to be adjusted.
⏰ Context from checks skipped due to timeout of 90000ms (22)
- GitHub Check: call-build-ecs-service-vulcan / (vulcan) Check docker image build
- GitHub Check: call-build-ecs-service-socks / (socks) Check docker image build
- GitHub Check: call-build-ecs-service-roundtable / (roundtable) Check docker image build
- GitHub Check: unit-end-to-end-and-integration
- GitHub Check: call-build-ecs-service-ender / (ender) Check docker image build
- GitHub Check: test-race
- GitHub Check: call-build-ecs-service-comlink / (comlink) Check docker image build
- GitHub Check: test / run_command
- GitHub Check: check-build-auxo
- GitHub Check: test-coverage-upload
- GitHub Check: golangci-lint
- GitHub Check: check-build-bazooka
- GitHub Check: liveness-test
- GitHub Check: lint
- GitHub Check: container-tests
- GitHub Check: benchmark
- GitHub Check: build
- GitHub Check: check-sample-pregenesis-up-to-date
- GitHub Check: Analyze (python)
- GitHub Check: Analyze (javascript-typescript)
- GitHub Check: Analyze (go)
- GitHub Check: Summary
🔇 Additional comments (1)
proto/dydxprotocol/clob/query.proto (1)
189-192: Verify filtering logic implementation.The field addition looks correct, but let's verify its integration with the existing streaming system.
Run the following to check for consistent handling across the codebase:
✅ Verification successful
Filtering logic is properly implemented
The field addition is well integrated with the existing streaming system. The implementation in
protocol/streaming/full_node_streaming_manager.gohandles the filtering of orderbook updates based on subaccount IDs, ensuring consistent behavior with the newfilter_ordersfield.🏁 Scripts executed
The following scripts were executed for the analysis:
Script:
#!/bin/bash # Description: Verify the integration of filter_orders field with streaming functionality # Check for corresponding handler implementations rg -g '*.go' -A 10 'StreamOrderbookUpdates.*Request.*filter.*orders' # Check for filtering logic in the streaming manager rg -g '*.go' -A 15 'func.*Subscribe.*bool.*filter' protocol/streaming/ # Look for tests verifying the filtering functionality rg -g '*_test.go' -A 10 'filter.*orders.*true' protocol/streaming/Length of output: 212
Script:
#!/bin/bash # First, find all orderbook and streaming related files fd -e go -e proto | rg -i "orderbook|streaming" # Then look for any filter-related code in these contexts rg -g '*.go' -A 5 'type.*Orderbook.*|func.*Orderbook' # Search for any streaming subscriber implementations rg -g '*.go' -A 5 'Subscribe.*Orderbook|Stream.*Orderbook' # Look for any filter-related logic in the codebase rg -g '*.go' -A 5 'filter.*[Oo]rder.*|[Oo]rder.*filter'Length of output: 75412
Changelist
Full Node Streaming provides an initial state and streaming updates for positions, orders, prices and fills.
The subscription API admits an optional sub account ID filter, which is only applied to the initial positions and position changes.
The user would like to apply the sub account ID filter to the order messages, in addition to position messages.
The change will add boolean flags to the Websocket and GRPC streaming API's:
For all endpoints, the previous behavior of not filtering orders for subaccounts is preserved by default.
If filtering orders is not specified, the code path remains the same for looping over stream updates.
If filtering orders is specified, the subscription's updates channel is replaced with a channel receiving
filtered updates from a new goroutine, which is only started in the event filtering orders is indicated.
Test Plan
Unit test
Author/Reviewer Checklist
state-breakinglabel.indexer-postgres-breakinglabel.PrepareProposalorProcessProposal, manually add the labelproposal-breaking.feature:[feature-name].backport/[branch-name].refactor,chore,bug.Summary by CodeRabbit
New Features
Improvements
Technical Updates
filterOrdersparameter.